package io.github.talelin.sleeve.module.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;

import java.util.Optional;

public class KafkaConsumer {

    @KafkaListener(id = "consumer-test",topics = "test") //@KafkaListener(id = "consumer-id7")那么当前消费者的消费组就是consumer-id7
    public void kafkaConsumer(ConsumerRecord<?, ?> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            String message = kafkaMessage.get().toString();
            System.out.println("Kafka消费成功 "+message);
        }
    }
}

